import asyncio
import json
from datetime import datetime, date
from typing import Optional, List, Dict, Any
import asyncpg
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
import os
from contextlib import asynccontextmanager
import functools

# 全局数据库连接池
db_pool = None

# 连接池重试配置
MAX_RETRY_ATTEMPTS = 3
RETRY_DELAY = 5  # 秒

# 数据库配置
DATABASE_CONFIG = {
    "host": "10.10.103.235",  # 请根据实际情况修改
    "port": 5432,
    "database": "chat_messages",  # 请根据实际情况修改
    "user": "xiaotangliexue",  # 请根据实际情况修改
    "password": "xiaotang"  # 请根据实际情况修改
}

# Pydantic模型
class MessageData(BaseModel):
    sourcedata: str
    td_role: str
    td_externalUserId: str
    corpid: str = ""
    groupId: str
    customerExternalUserId: str
    contactType: int
    source: int
    td_userNickNames: List[str]
    mtype: str = ""
    datetime: str
    td_userContactName: str
    td_botUserId: str
    imContactId: str
    messageType: int
    payload: Dict[str, Any]
    mlevel: str = ""
    sourcetype: str
    m1reason: str = ""
    coworker: Optional[bool] = None
    contactName: str
    messageId: str
    m2status: int
    externalUserId: str
    mnotes: str = ""
    imBotId: str
    imRoomId: str = ""
    m1notes: str = ""
    mreason: str = ""
    isSelf: Optional[bool] = None
    chatId: str
    m2level: str = ""
    mstatus: int
    td_imContactId: str
    roomTopic: str = ""
    botUserId: str
    botId: str
    timestamp: int
    c_corpid: str
    utime: int
    m1level: str = ""
    m2notes: str = ""
    m2type: str = ""
    roomWecomChatId: str = ""
    m1status: int
    avatar: str = ""
    textMd5: str = ""
    m2reason: str = ""
    m1type: str = ""
    td_imBotId: str
    td_contactName: str

class MessageBatch(BaseModel):
    messages: List[MessageData]

# 数据库连接函数
async def init_database():
    """初始化数据库连接池和创建必要的表"""
    global db_pool
    
    try:
        # 创建连接池 - 优化并发性能配置
        db_pool = await asyncpg.create_pool(
            host=DATABASE_CONFIG["host"],
            port=DATABASE_CONFIG["port"],
            database=DATABASE_CONFIG["database"],
            user=DATABASE_CONFIG["user"],
            password=DATABASE_CONFIG["password"],
            min_size=10,  # 增加最小连接数以支持更好的并发
            max_size=50,  # 增加最大连接数以处理高并发场景
            max_queries=50000,  # 每个连接的最大查询数
            max_inactive_connection_lifetime=300.0,  # 非活跃连接的最大生存时间（秒）
            command_timeout=60,  # 命令超时时间
            server_settings={
                'application_name': '聊天消息管理系统',
                'jit': 'off'  # 关闭JIT以提高稳定性
            }
        )
        
        # 创建基础表结构（如果不存在）
        async with db_pool.acquire() as conn:
            # 创建chat_messages表（用于存储完整的聊天消息）
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS chat_messages (
                    id SERIAL PRIMARY KEY,
                    customer_external_user_id VARCHAR(255),
                    group_id VARCHAR(255),
                    td_role VARCHAR(50),
                    td_bot_user_id VARCHAR(255),
                    payload JSONB,
                    datetime TIMESTAMP,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    table_date DATE
                )
            """)
            
            # 创建索引以提高查询性能
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_chat_messages_user_id 
                ON chat_messages(customer_external_user_id)
            """)
            
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_chat_messages_datetime 
                ON chat_messages(datetime)
            """)
            
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_chat_messages_role 
                ON chat_messages(td_role)
            """)
            
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_chat_messages_payload_gin 
                ON chat_messages USING GIN (payload)
            """)
            
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_chat_messages_table_date 
                ON chat_messages(table_date)
            """)
            
        # 验证连接池是否正常工作
        async with db_pool.acquire() as test_conn:
            await test_conn.fetchval('SELECT 1')
        
        print(f"数据库连接池初始化完成 - 最小连接数: {db_pool._minsize}, 最大连接数: {db_pool._maxsize}")
        
    except Exception as e:
        print(f"数据库初始化失败: {e}")
        if db_pool:
            await db_pool.close()
            db_pool = None
        raise

async def ensure_db_connection():
    """确保数据库连接可用，如果连接断开则尝试重连"""
    global db_pool
    
    if not db_pool or db_pool._closed:
        print("检测到数据库连接池不可用，尝试重新初始化...")
        for attempt in range(MAX_RETRY_ATTEMPTS):
            try:
                await init_database()
                print(f"数据库连接池重新初始化成功 (尝试 {attempt + 1}/{MAX_RETRY_ATTEMPTS})")
                return True
            except Exception as e:
                print(f"数据库重连失败 (尝试 {attempt + 1}/{MAX_RETRY_ATTEMPTS}): {e}")
                if attempt < MAX_RETRY_ATTEMPTS - 1:
                    await asyncio.sleep(RETRY_DELAY)
                else:
                    print("数据库重连失败，已达到最大重试次数")
                    return False
    return True

def with_db_retry(func):
    """数据库操作重试装饰器"""
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        if not await ensure_db_connection():
            raise HTTPException(status_code=503, detail="数据库服务不可用")
        
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            # 如果是连接相关错误，尝试重连一次
            if "connection" in str(e).lower() or "pool" in str(e).lower():
                print(f"检测到连接错误，尝试重连: {e}")
                if await ensure_db_connection():
                    return await func(*args, **kwargs)
            raise e
    return wrapper

async def get_db_pool_status():
    """获取数据库连接池状态信息"""
    if not db_pool:
        return {"status": "未初始化", "active_connections": 0, "idle_connections": 0}
    
    return {
        "status": "正常" if not db_pool._closed else "已关闭",
        "min_size": db_pool._minsize,
        "max_size": db_pool._maxsize,
        "current_size": db_pool.get_size(),
        "idle_connections": db_pool.get_idle_size(),
        "active_connections": db_pool.get_size() - db_pool.get_idle_size()
    }

async def create_daily_summary_table(table_date: str):
    """创建每日统计表"""
    table_name = f"day_{table_date.replace('-', '')}"
    
    async with db_pool.acquire() as conn:
        await conn.execute(f"""
            CREATE TABLE IF NOT EXISTS {table_name} (
                id SERIAL PRIMARY KEY,
                customer_external_user_id VARCHAR(255),
                group_id VARCHAR(255),
                role_user INTEGER DEFAULT 0,
                role_boter INTEGER DEFAULT 0,
                sum_messages INTEGER DEFAULT 0,
                datetime TIMESTAMP,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(customer_external_user_id, group_id)
            )
        """)
        
        # 创建索引
        await conn.execute(f"""
            CREATE INDEX IF NOT EXISTS idx_{table_name}_user_id 
            ON {table_name}(customer_external_user_id)
        """)
        
        # 为datetime创建索引
        await conn.execute(f"""
            CREATE INDEX IF NOT EXISTS idx_{table_name}_datetime 
            ON {table_name}(datetime)
        """)

# FastAPI应用
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时初始化数据库
    print("正在启动聊天消息管理系统...")
    await init_database()
    print("系统启动完成，数据库连接池已就绪")
    
    try:
        yield
    finally:
        # 关闭时清理资源
        print("正在关闭系统...")
        if db_pool and not db_pool._closed:
            print("正在关闭数据库连接池...")
            await db_pool.close()
            print("数据库连接池已关闭")
        print("系统已安全关闭")

app = FastAPI(lifespan=lifespan, title="聊天消息管理系统", version="1.0.0")

# 1. 查询接口
@app.get("/v1/query_messages")
@with_db_retry
async def query_messages(
    customerexternaluserid: Optional[str] = Query(None, description="用户ID"),
    datetime_before: Optional[str] = Query(None, description="查询此时间之前的消息"),
    datetime_after: Optional[str] = Query(None, description="查询此时间之后的消息"),
    top: Optional[int] = Query(None, description="返回发送消息最多的前N个用户"),
    text_search: Optional[str] = Query(None, description="搜索消息文本内容"),
    has_voice: Optional[bool] = Query(None, description="是否包含语音消息"),
    fuzzy_match: Optional[bool] = Query(False, description="是否启用模糊匹配用户ID")
):
    """查询聊天消息"""
    try:
        async with db_pool.acquire() as conn:
            # 构建查询条件
            conditions = []
            params = []
            param_count = 0
            
            if customerexternaluserid:
                param_count += 1
                if fuzzy_match:
                    conditions.append(f"customer_external_user_id LIKE ${param_count}")
                    params.append(f"%{customerexternaluserid}%")
                else:
                    conditions.append(f"customer_external_user_id = ${param_count}")
                    params.append(customerexternaluserid)
            
            if datetime_before:
                try:
                    # 尝试解析日期时间字符串
                    if 'T' in datetime_before:
                        # ISO格式：2025-08-09T22:45:50
                        parsed_datetime = datetime.fromisoformat(datetime_before.replace('Z', '+00:00'))
                    else:
                        # 空格分隔格式：2025-08-09 22:45:50
                        parsed_datetime = datetime.strptime(datetime_before, '%Y-%m-%d %H:%M:%S')
                    
                    param_count += 1
                    conditions.append(f"datetime < ${param_count}")
                    params.append(parsed_datetime)
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=f"Invalid datetime_before format: {datetime_before}. Expected format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DDTHH:MM:SS'")
            
            if datetime_after:
                try:
                    # 尝试解析日期时间字符串
                    if 'T' in datetime_after:
                        # ISO格式：2025-08-09T22:45:50
                        parsed_datetime = datetime.fromisoformat(datetime_after.replace('Z', '+00:00'))
                    else:
                        # 空格分隔格式：2025-08-09 22:45:50
                        parsed_datetime = datetime.strptime(datetime_after, '%Y-%m-%d %H:%M:%S')
                    
                    param_count += 1
                    conditions.append(f"datetime > ${param_count}")
                    params.append(parsed_datetime)
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=f"Invalid datetime_after format: {datetime_after}. Expected format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DDTHH:MM:SS'")
            
            if text_search:
                param_count += 1
                conditions.append(f"payload->>'text' ILIKE ${param_count}")
                params.append(f"%{text_search}%")
            
            if has_voice is not None:
                if has_voice:
                    conditions.append("payload ? 'voiceUrl'")
                else:
                    conditions.append("NOT (payload ? 'voiceUrl')")
            
            # 构建基础查询
            base_query = "SELECT * FROM chat_messages"
            if conditions:
                base_query += " WHERE " + " AND ".join(conditions)
            
            # 如果需要按消息数量排序用户
            if top:
                if customerexternaluserid:
                    # 查询特定用户的消息，按时间排序
                    query = base_query + f" ORDER BY datetime DESC LIMIT {top}"
                else:
                    # 基于day_表中role_user的最大值进行排序
                    
                    # 首先获取所有day_开头的表名
                    day_tables_query = """
                        SELECT table_name 
                        FROM information_schema.tables 
                        WHERE table_schema = 'public' 
                        AND table_name LIKE 'day_%'
                    """
                    day_tables = await conn.fetch(day_tables_query)
                    
                    if not day_tables:
                        # 如果没有day_表，回退到原来的逻辑
                        query = f"""
                            WITH user_total_stats AS (
                                SELECT 
                                    customer_external_user_id,
                                    COUNT(*) as total_message_count
                                FROM chat_messages 
                                GROUP BY customer_external_user_id
                            )
                            SELECT 
                                customer_external_user_id,
                                total_message_count,
                                NULL as peak_date,
                                NULL as peak_daily_count,
                                0 as role_user
                            FROM user_total_stats
                            ORDER BY total_message_count DESC
                            LIMIT {top}
                        """
                    else:
                        # 构建UNION查询来获取所有day_表中的最大role_user值
                        union_queries = []
                        for table in day_tables:
                            table_name = table['table_name']
                            union_queries.append(f"""
                                SELECT 
                                    customer_external_user_id,
                                    role_user,
                                    '{table_name}' as source_table
                                FROM {table_name}
                            """)
                        
                        union_query = " UNION ALL ".join(union_queries)
                        
                        query = f"""
                             WITH all_day_data AS (
                                 {union_query}
                             ),
                             user_max_role_with_date AS (
                                 SELECT DISTINCT ON (customer_external_user_id)
                                     customer_external_user_id,
                                     role_user as max_role_user,
                                     source_table
                                 FROM all_day_data
                                 ORDER BY customer_external_user_id, role_user DESC, source_table DESC
                             ),
                             ranked_users AS (
                                 SELECT 
                                     customer_external_user_id,
                                     max_role_user,
                                     source_table,
                                     ROW_NUMBER() OVER (ORDER BY max_role_user DESC, customer_external_user_id) as rank
                                 FROM user_max_role_with_date
                             ),
                             top_users AS (
                                 SELECT 
                                     customer_external_user_id,
                                     max_role_user,
                                     source_table
                                 FROM ranked_users
                                 WHERE rank <= {top}
                             ),
                             user_total_stats AS (
                                 SELECT 
                                     customer_external_user_id,
                                     COUNT(*) as total_message_count
                                 FROM chat_messages 
                                 GROUP BY customer_external_user_id
                             ),
                             user_daily_stats AS (
                                 SELECT 
                                     customer_external_user_id,
                                     DATE(datetime) as message_date,
                                     COUNT(*) as daily_count
                                 FROM chat_messages 
                                 GROUP BY customer_external_user_id, DATE(datetime)
                             ),
                             user_peak_day AS (
                                 SELECT 
                                     uds.customer_external_user_id,
                                     uds.message_date as peak_date,
                                     uds.daily_count as peak_daily_count,
                                     ROW_NUMBER() OVER (PARTITION BY uds.customer_external_user_id ORDER BY uds.daily_count DESC, uds.message_date DESC) as rn
                                 FROM user_daily_stats uds
                             )
                             SELECT 
                                 tu.customer_external_user_id,
                                 COALESCE(uts.total_message_count, 0) as total_message_count,
                                 upd.peak_date,
                                 upd.peak_daily_count,
                                 tu.max_role_user as role_user,
                                 REPLACE(tu.source_table, 'day_', '') as role_user_peak_date
                             FROM top_users tu
                             LEFT JOIN user_total_stats uts ON tu.customer_external_user_id = uts.customer_external_user_id
                             LEFT JOIN user_peak_day upd ON tu.customer_external_user_id = upd.customer_external_user_id AND upd.rn = 1
                             ORDER BY tu.max_role_user DESC, tu.customer_external_user_id
                         """
            else:
                query = base_query + " ORDER BY datetime DESC"
            
            # 执行查询
            rows = await conn.fetch(query, *params)
            
            # 转换结果
            results = []
            for row in rows:
                result = dict(row)
                # 确保datetime字段格式正确
                if 'datetime' in result and result['datetime']:
                    result['datetime'] = result['datetime'].isoformat()
                if 'created_at' in result and result['created_at']:
                    result['created_at'] = result['created_at'].isoformat()
                results.append(result)
            
            return {
                "success": True,
                "data": results,
                "count": len(results)
            }
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"查询失败: {str(e)}")

# 2. 存储接口
@app.post("/v1/server_messages")
@with_db_retry
async def server_messages(message_batch: MessageBatch):
    """存储聊天消息并更新每日统计"""
    try:
        async with db_pool.acquire() as conn:
            async with conn.transaction():
                for message_data in message_batch.messages:
                    # 解析日期
                    msg_datetime = datetime.strptime(message_data.datetime, "%Y-%m-%d %H:%M:%S")
                    table_date = msg_datetime.date()
                    table_date_str = table_date.strftime("%Y%m%d")
                    
                    # 创建每日统计表（如果不存在）
                    await create_daily_summary_table(table_date.strftime("%Y-%m-%d"))
                    
                    # 插入完整消息数据到chat_messages表
                    await conn.execute("""
                        INSERT INTO chat_messages (
                            customer_external_user_id, group_id, td_role, td_bot_user_id,
                            payload, datetime, table_date
                        ) VALUES ($1, $2, $3, $4, $5, $6, $7)
                    """, 
                        message_data.customerExternalUserId,
                        message_data.groupId,
                        message_data.td_role,
                        message_data.td_botUserId,
                        json.dumps(message_data.payload),
                        msg_datetime,
                        table_date
                    )
                    
                    # 更新每日统计表
                    daily_table = f"day_{table_date_str}"
                    
                    # 检查记录是否存在
                    existing = await conn.fetchrow(f"""
                        SELECT role_user, role_boter FROM {daily_table}
                        WHERE customer_external_user_id = $1 AND group_id = $2
                    """, message_data.customerExternalUserId, message_data.groupId)
                    
                    if existing:
                        # 更新现有记录
                        if message_data.td_role == "user":
                            new_user_count = existing['role_user'] + 1
                            new_boter_count = existing['role_boter']
                        elif message_data.td_role == "boter":
                            new_user_count = existing['role_user']
                            new_boter_count = existing['role_boter'] + 1
                        else:
                            # 如果角色不是user或boter，保持原有计数不变
                            new_user_count = existing['role_user']
                            new_boter_count = existing['role_boter']
                        
                        await conn.execute(f"""
                            UPDATE {daily_table} SET
                                role_user = $1,
                                role_boter = $2,
                                sum_messages = $3,
                                datetime = $4,
                                updated_at = CURRENT_TIMESTAMP
                            WHERE customer_external_user_id = $5 AND group_id = $6
                        """, new_user_count, new_boter_count, 
                            new_user_count + new_boter_count,
                            msg_datetime,
                            message_data.customerExternalUserId, message_data.groupId)
                    else:
                        # 插入新记录
                        role_user = 1 if message_data.td_role == "user" else 0
                        role_boter = 1 if message_data.td_role == "boter" else 0
                        
                        await conn.execute(f"""
                            INSERT INTO {daily_table} (
                                customer_external_user_id, group_id, role_user, role_boter, sum_messages,
                                datetime
                            ) VALUES ($1, $2, $3, $4, $5, $6)
                        """, message_data.customerExternalUserId, message_data.groupId,
                            role_user, role_boter, role_user + role_boter,
                            msg_datetime)
        
        return {
            "success": True,
            "message": f"成功处理 {len(message_batch.messages)} 条消息",
            "processed_count": len(message_batch.messages)
        }
        
    except Exception as e:
        import traceback
        error_msg = f"存储失败: {str(e)}"
        print(f"ERROR in server_messages: {error_msg}")
        print(f"Traceback: {traceback.format_exc()}")
        raise HTTPException(status_code=500, detail=error_msg)



# 3. 获取用户完整对话记录接口
@app.get("/v1/user_conversation/{customer_external_user_id}")
async def get_user_conversation(
    customer_external_user_id: str,
    date: Optional[str] = Query(None, description="指定日期，格式：YYYY-MM-DD"),
    group_id: Optional[str] = Query(None, description="群组ID"),
    limit: Optional[int] = Query(100, description="返回消息数量限制")
):
    """获取用户完整对话记录，按时间顺序返回用户和客服的消息"""
    try:
        async with db_pool.acquire() as conn:
            # 构建查询条件
            conditions = ["customer_external_user_id = $1"]
            params = [customer_external_user_id]
            param_count = 1
            
            if date:
                param_count += 1
                conditions.append(f"DATE(datetime) = ${param_count}")
                params.append(date)
            
            if group_id:
                param_count += 1
                conditions.append(f"group_id = ${param_count}")
                params.append(group_id)
            
            # 查询消息记录，按时间顺序排序
            query = f"""
                SELECT 
                    td_role,
                    payload,
                    datetime,
                    group_id,
                    message_id
                FROM chat_messages 
                WHERE {' AND '.join(conditions)}
                ORDER BY datetime ASC
                LIMIT ${param_count + 1}
            """
            params.append(limit)
            
            rows = await conn.fetch(query, *params)
            
            # 格式化对话记录
            conversation = []
            for row in rows:
                # 解析payload获取消息内容
                payload = json.loads(row['payload']) if isinstance(row['payload'], str) else row['payload']
                
                # 获取消息文本内容
                message_text = payload.get('text', '')
                
                # 如果没有文本内容，检查是否有其他类型的消息
                if not message_text:
                    if payload.get('voiceUrl'):
                        message_text = f"[语音消息，时长：{payload.get('duration', 0)}秒]"
                    elif payload.get('imageUrl'):
                        message_text = "[图片消息]"
                    elif payload.get('fileUrl'):
                        message_text = "[文件消息]"
                    else:
                        message_text = "[其他类型消息]"
                
                # 格式化时间
                formatted_time = row['datetime'].strftime("%Y-%m-%d %H:%M:%S")
                
                # 确定角色显示名称
                role_display = "客服" if row['td_role'] == "boter" else "user"
                
                conversation.append({
                    "role": role_display,
                    "message": message_text,
                    "datetime": formatted_time,
                    "formatted_display": f"{role_display}：{message_text}（{formatted_time}）",
                    "group_id": row['group_id'],
                    "message_id": row['message_id'],
                    "payload": payload
                })
            
            return {
                "success": True,
                "customer_external_user_id": customer_external_user_id,
                "conversation": conversation,
                "total_messages": len(conversation),
                "query_params": {
                    "date": date,
                    "group_id": group_id,
                    "limit": limit
                }
            }
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取对话记录失败: {str(e)}")

# 4. 获取用户对话统计信息接口
@app.get("/v1/user_conversation_stats/{customer_external_user_id}")
@with_db_retry
async def get_user_conversation_stats(
    customer_external_user_id: str,
    date: Optional[str] = Query(None, description="指定日期，格式：YYYY-MM-DD")
):
    """获取用户对话统计信息"""
    try:
        async with db_pool.acquire() as conn:
            # 构建查询条件
            conditions = ["customer_external_user_id = $1"]
            params = [customer_external_user_id]
            param_count = 1
            
            if date:
                param_count += 1
                conditions.append(f"DATE(datetime) = ${param_count}")
                params.append(date)
            
            # 查询统计信息
            stats_query = f"""
                SELECT 
                    td_role,
                    COUNT(*) as message_count,
                    MIN(datetime) as first_message_time,
                    MAX(datetime) as last_message_time
                FROM chat_messages 
                WHERE {' AND '.join(conditions)}
                GROUP BY td_role
            """
            
            stats_rows = await conn.fetch(stats_query, *params)
            
            # 整理统计数据
            stats = {
                "user_messages": 0,
                "boter_messages": 0,
                "total_messages": 0,
                "first_message_time": None,
                "last_message_time": None,
                "conversation_duration": None
            }
            
            all_times = []
            for row in stats_rows:
                if row['td_role'] == 'user':
                    stats['user_messages'] = row['message_count']
                elif row['td_role'] == 'boter':
                    stats['boter_messages'] = row['message_count']
                
                all_times.extend([row['first_message_time'], row['last_message_time']])
            
            stats['total_messages'] = stats['user_messages'] + stats['boter_messages']
            
            if all_times:
                all_times = [t for t in all_times if t is not None]
                if all_times:
                    stats['first_message_time'] = min(all_times).strftime("%Y-%m-%d %H:%M:%S")
                    stats['last_message_time'] = max(all_times).strftime("%Y-%m-%d %H:%M:%S")
                    
                    # 计算对话持续时间（分钟）
                    duration = max(all_times) - min(all_times)
                    stats['conversation_duration'] = int(duration.total_seconds() / 60)
            
            return {
                "success": True,
                "customer_external_user_id": customer_external_user_id,
                "stats": stats,
                "query_params": {
                    "date": date
                }
            }
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取对话统计失败: {str(e)}")

@app.get("/v1/chat_messages")
@with_db_retry
async def get_chat_messages(
    customerExternalUserId: str = Query(..., description="用户ID"),
    groupId: Optional[str] = Query(None, description="群组ID"),
    datetime_after: Optional[str] = Query(None, description="查询此时间之后的消息"),
    datetime_before: Optional[str] = Query(None, description="查询此时间之前的消息"),
    td_bot_user_id: Optional[str] = Query(None, description="机器人用户ID"),
    debug: Optional[bool] = Query(False, description="是否返回调试信息")
):
    """获取指定用户和群组的聊天消息，按时间顺序返回格式化的对话内容"""
    try:
        async with db_pool.acquire() as conn:
            # 构建查询条件
            conditions = ["customer_external_user_id = $1"]
            params = [customerExternalUserId]
            param_count = 1
            
            # 如果提供了groupId，则添加到查询条件
            if groupId:
                param_count += 1
                conditions.append(f"group_id = ${param_count}")
                params.append(groupId)
            
            # 如果提供了td_bot_user_id，则添加到查询条件
            if td_bot_user_id:
                param_count += 1
                conditions.append(f"td_bot_user_id = ${param_count}")
                params.append(td_bot_user_id)
            
            debug_info = {
                "customerExternalUserId": customerExternalUserId,
                "groupId": groupId,
                "datetime_after": datetime_after,
                "datetime_before": datetime_before,
                "td_bot_user_id": td_bot_user_id,
                "conditions": conditions.copy(),
                "params": params.copy()
            }
            
            if datetime_before:
                try:
                    # 尝试解析日期时间字符串
                    if 'T' in datetime_before:
                        # ISO格式：2025-08-09T22:45:50
                        parsed_datetime = datetime.fromisoformat(datetime_before.replace('Z', '+00:00'))
                    else:
                        # 空格分隔格式：2025-08-09 22:45:50
                        parsed_datetime = datetime.strptime(datetime_before, '%Y-%m-%d %H:%M:%S')
                    
                    param_count += 1
                    conditions.append(f"datetime < ${param_count}")
                    params.append(parsed_datetime)
                    debug_info["parsed_datetime_before"] = parsed_datetime.isoformat()
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=f"Invalid datetime_before format: {datetime_before}. Expected format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DDTHH:MM:SS'")
            
            if datetime_after:
                try:
                    # 尝试解析日期时间字符串
                    if 'T' in datetime_after:
                        # ISO格式：2025-08-09T22:45:50
                        parsed_datetime = datetime.fromisoformat(datetime_after.replace('Z', '+00:00'))
                    else:
                        # 空格分隔格式：2025-08-09 22:45:50
                        parsed_datetime = datetime.strptime(datetime_after, '%Y-%m-%d %H:%M:%S')
                    
                    param_count += 1
                    conditions.append(f"datetime > ${param_count}")
                    params.append(parsed_datetime)
                    debug_info["parsed_datetime_after"] = parsed_datetime.isoformat()
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=f"Invalid datetime_after format: {datetime_after}. Expected format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DDTHH:MM:SS'")
            
            # 查询消息记录，按时间顺序排序
            query = f"""
                SELECT 
                    td_role,
                    payload,
                    datetime
                FROM chat_messages 
                WHERE {' AND '.join(conditions)}
                ORDER BY datetime ASC
            """
            
            debug_info["final_query"] = query
            debug_info["final_params"] = params
            
            rows = await conn.fetch(query, *params)
            debug_info["total_rows_found"] = len(rows)
            debug_info["sample_rows"] = []
            
            # 记录前几行数据用于调试
            for i, row in enumerate(rows[:5]):
                debug_info["sample_rows"].append({
                    "td_role": row['td_role'],
                    "datetime": row['datetime'].isoformat(),
                    "payload_preview": str(row['payload'])[:100] + "..." if len(str(row['payload'])) > 100 else str(row['payload'])
                })
            
            # 构建返回结果
            result = {
                "customerExternalUserId": customerExternalUserId,
                "groupId": groupId,
                "td_bot_user_id": td_bot_user_id,
                "messages": []
            }
            
            # 按顺序添加消息
            message_count = 0
            for row in rows:
                # 解析payload获取消息内容
                payload = json.loads(row['payload']) if isinstance(row['payload'], str) else row['payload']
                
                # 获取消息文本内容
                message_text = payload.get('text', '')
                
                # 如果没有文本内容，检查是否有其他类型的消息
                if not message_text:
                    if payload.get('voiceUrl'):
                        message_text = f"[语音消息，时长：{payload.get('duration', 0)}秒]"
                    elif payload.get('imageUrl'):
                        message_text = "[图片消息]"
                    elif payload.get('fileUrl'):
                        message_text = "[文件消息]"
                    else:
                        message_text = "[其他类型消息]"
                
                # 格式化时间
                formatted_time = row['datetime'].strftime("%Y-%m-%d %H:%M:%S")
                
                # 根据角色确定字段名
                role_field = "user" if row['td_role'] == "user" else "boter"
                
                # 添加到消息列表中
                message_entry = {
                    "role": role_field,
                    "content": f"{message_text}（{formatted_time}）"
                }
                result["messages"].append(message_entry)
                
                # 同时也添加为单独的字段（保持原有格式）
                result[role_field] = f"{message_text}（{formatted_time}）"
                message_count += 1
            
            debug_info["messages_processed"] = message_count
            
            # 如果启用调试模式，返回调试信息
            if debug:
                result["_debug"] = debug_info
            
            return result
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取聊天消息失败: {str(e)}")

@app.get("/v1/download_chat_messages")
@with_db_retry
async def download_chat_messages(
    customerExternalUserId: str = Query(..., description="用户ID"),
    groupId: Optional[str] = Query(None, description="群组ID"),
    datetime_after: Optional[str] = Query(None, description="查询此时间之后的消息"),
    datetime_before: Optional[str] = Query(None, description="查询此时间之前的消息"),
    debug: Optional[bool] = Query(False, description="是否返回调试信息")
):
    """下载指定用户和群组的聊天消息，返回JSON格式的对话数据"""
    try:
        async with db_pool.acquire() as conn:
            # 构建查询条件
            conditions = ["customer_external_user_id = $1"]
            params = [customerExternalUserId]
            param_count = 1
            
            # 如果提供了groupId，则添加到查询条件
            if groupId:
                param_count += 1
                conditions.append(f"group_id = ${param_count}")
                params.append(groupId)
            
            debug_info = {
                "customerExternalUserId": customerExternalUserId,
                "groupId": groupId,
                "datetime_after": datetime_after,
                "datetime_before": datetime_before,
                "conditions": conditions.copy(),
                "params": params.copy()
            }
            
            if datetime_before:
                try:
                    # 尝试解析日期时间字符串
                    if 'T' in datetime_before:
                        # ISO格式：2025-08-09T22:45:50
                        parsed_datetime = datetime.fromisoformat(datetime_before.replace('Z', '+00:00'))
                    else:
                        # 空格分隔格式：2025-08-09 22:45:50
                        parsed_datetime = datetime.strptime(datetime_before, '%Y-%m-%d %H:%M:%S')
                    
                    param_count += 1
                    conditions.append(f"datetime < ${param_count}")
                    params.append(parsed_datetime)
                    debug_info["parsed_datetime_before"] = parsed_datetime.isoformat()
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=f"Invalid datetime_before format: {datetime_before}. Expected format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DDTHH:MM:SS'")
            
            if datetime_after:
                try:
                    # 尝试解析日期时间字符串
                    if 'T' in datetime_after:
                        # ISO格式：2025-08-09T22:45:50
                        parsed_datetime = datetime.fromisoformat(datetime_after.replace('Z', '+00:00'))
                    else:
                        # 空格分隔格式：2025-08-09 22:45:50
                        parsed_datetime = datetime.strptime(datetime_after, '%Y-%m-%d %H:%M:%S')
                    
                    param_count += 1
                    conditions.append(f"datetime > ${param_count}")
                    params.append(parsed_datetime)
                    debug_info["parsed_datetime_after"] = parsed_datetime.isoformat()
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=f"Invalid datetime_after format: {datetime_after}. Expected format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DDTHH:MM:SS'")
            
            # 查询消息记录，按时间顺序排序
            query = f"""
                SELECT 
                    td_role,
                    payload,
                    datetime
                FROM chat_messages 
                WHERE {' AND '.join(conditions)}
                ORDER BY datetime ASC
            """
            
            debug_info["final_query"] = query
            debug_info["final_params"] = params
            
            rows = await conn.fetch(query, *params)
            debug_info["total_rows_found"] = len(rows)
            debug_info["sample_rows"] = []
            
            # 记录前几行数据用于调试
            for i, row in enumerate(rows[:5]):
                debug_info["sample_rows"].append({
                    "td_role": row['td_role'],
                    "datetime": row['datetime'].isoformat(),
                    "payload_preview": str(row['payload'])[:100] + "..." if len(str(row['payload'])) > 100 else str(row['payload'])
                })
            
            # 构建返回结果 - 与chat_messages路由保持一致的结构
            result = {
                "customerExternalUserId": customerExternalUserId,
                "groupId": groupId,
                "messages": [],
                "download_info": {
                    "total_messages": 0,
                    "download_time": datetime.now().isoformat(),
                    "query_time_range": {
                        "after": datetime_after,
                        "before": datetime_before
                    }
                }
            }
            
            # 按顺序添加消息
            message_count = 0
            for row in rows:
                # 解析payload获取消息内容
                payload = json.loads(row['payload']) if isinstance(row['payload'], str) else row['payload']
                
                # 获取消息文本内容
                message_text = payload.get('text', '')
                
                # 如果没有文本内容，检查是否有其他类型的消息
                if not message_text:
                    if payload.get('voiceUrl'):
                        message_text = f"[语音消息，时长：{payload.get('duration', 0)}秒]"
                    elif payload.get('imageUrl'):
                        message_text = "[图片消息]"
                    elif payload.get('fileUrl'):
                        message_text = "[文件消息]"
                    else:
                        message_text = "[其他类型消息]"
                
                # 格式化时间
                formatted_time = row['datetime'].strftime("%Y-%m-%d %H:%M:%S")
                
                # 根据角色确定字段名
                role_field = "user" if row['td_role'] == "user" else "boter"
                
                # 添加到消息列表中
                message_entry = {
                    "role": role_field,
                    "content": f"{message_text}（{formatted_time}）",
                    "raw_content": message_text,
                    "timestamp": formatted_time,
                    "message_type": "text" if payload.get('text') else (
                        "voice" if payload.get('voiceUrl') else (
                            "image" if payload.get('imageUrl') else (
                                "file" if payload.get('fileUrl') else "other"
                            )
                        )
                    )
                }
                result["messages"].append(message_entry)
                
                # 同时也添加为单独的字段（保持原有格式）
                result[role_field] = f"{message_text}（{formatted_time}）"
                message_count += 1
            
            # 更新下载信息
            result["download_info"]["total_messages"] = message_count
            
            debug_info["messages_processed"] = message_count
            
            # 如果启用调试模式，返回调试信息
            if debug:
                result["_debug"] = debug_info
            
            return result
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"下载聊天消息失败: {str(e)}")

@app.get("/debug/check_user_data/{user_id}")
async def debug_check_user_data(user_id: str):
    """调试接口：检查用户数据"""
    try:
        async with db_pool.acquire() as conn:
            # 检查chat_messages表
            chat_count = await conn.fetchval(
                "SELECT COUNT(*) FROM chat_messages WHERE customer_external_user_id = $1", 
                user_id
            )
            
            # 获取chat_messages表中的样本数据
            chat_samples = await conn.fetch(
                "SELECT customer_external_user_id, td_role, datetime FROM chat_messages WHERE customer_external_user_id = $1 LIMIT 5", 
                user_id
            )
            
            # 检查所有day_开头的表
            day_tables = await conn.fetch(
                "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name LIKE 'day_%'"
            )
            
            day_data = {}
            for table_row in day_tables:
                table_name = table_row['table_name']
                try:
                    count = await conn.fetchval(
                        f"SELECT COUNT(*) FROM {table_name} WHERE customer_external_user_id = $1", 
                        user_id
                    )
                    day_data[table_name] = count
                except Exception as e:
                    day_data[table_name] = f"Error: {str(e)}"
            
            return {
                "user_id": user_id,
                "chat_messages_count": chat_count,
                "chat_samples": [dict(row) for row in chat_samples],
                "day_tables_data": day_data,
                "total_day_tables": len(day_tables)
            }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Debug check failed: {str(e)}")









# 健康检查接口
@app.get("/health")
async def health_check():
    """系统健康检查"""
    try:
        # 检查数据库连接
        async with db_pool.acquire() as conn:
            db_version = await conn.fetchval("SELECT version()")
            
        # 获取连接池状态
        pool_status = await get_db_pool_status()
        
        return {
            "status": "healthy",
            "timestamp": datetime.now().isoformat(),
            "database": {
                "status": "connected",
                "version": db_version.split()[0:2]  # 只返回PostgreSQL版本信息
            },
            "connection_pool": pool_status,
            "system": {
                "title": "聊天消息管理系统",
                "version": "1.0.0"
            }
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "timestamp": datetime.now().isoformat(),
            "database": {"status": "disconnected", "error": str(e)},
            "connection_pool": await get_db_pool_status()
        }

@app.get("/health/db-pool")
async def db_pool_status():
    """数据库连接池状态监控"""
    try:
        pool_status = await get_db_pool_status()
        
        # 测试连接池性能
        start_time = datetime.now()
        async with db_pool.acquire() as conn:
            await conn.fetchval("SELECT 1")
        response_time = (datetime.now() - start_time).total_seconds() * 1000
        
        return {
            "status": "success",
            "timestamp": datetime.now().isoformat(),
            "pool_info": pool_status,
            "performance": {
                "connection_acquire_time_ms": round(response_time, 2)
            }
        }
    except Exception as e:
        return {
            "status": "error",
            "timestamp": datetime.now().isoformat(),
            "error": str(e),
            "pool_info": await get_db_pool_status()
        }

if __name__ == "__main__":
    import uvicorn
    # 优化启动配置以支持高并发
    uvicorn.run(
        "demo:app", 
        host="0.0.0.0", 
        port=8585, 
        reload=True,
        workers=1,  # 开发环境使用单进程，生产环境建议根据CPU核心数调整
        loop="asyncio",  # 使用asyncio事件循环
        access_log=True,
        log_level="info"
    )
